package cn.tedu.micro.demo.client.delay;

import cn.tedu.micro.demo.client.RabbitChannelManager;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * 测试声明延迟队列结构
 */
@Slf4j
public class DelayDemo {
    //声明一个可以处理业务队列产生死信的死信队列结构
    /**
     * 参考约定名称和绑定关系 声明一套死信的组件结构
     */
    private static final String DELAY_EX="delay_ex";
    private static final String DELAY_Q="delay_q";
    private static final String DLX_EX="dlx_ex";
    private static final String DLX_Q="dlx_q";
    private static final String DLX_RK="dlx_rk";
    @Test
    public void delayDeclare() throws IOException {
        //1.获取连接
        Channel channel = RabbitChannelManager.newChannel();
        //删除原有交换机和队列
        channel.exchangeDelete(DELAY_EX);
        channel.queueDelete(DELAY_Q);
        channel.exchangeDelete(DLX_EX);
        channel.queueDelete(DLX_Q);
        //2.声明死信交换机 死信队列 绑定关系
        channel.exchangeDeclare(DLX_EX, BuiltinExchangeType.DIRECT);
        channel.queueDeclare(DLX_Q, false, false, false, null);
        channel.queueBind(DLX_Q, DLX_EX, DLX_RK);
        //3.声明业务交换机 业务队列 业务队列设置死信的数据属性
        channel.exchangeDeclare(DELAY_EX, BuiltinExchangeType.FANOUT);
        //3.1声明业务队列,只要队列出现死信,就要携带死信路由进入死信交换机
        Map<String,Object> args= new HashMap<>();
        args.put("x-dead-letter-exchange",DLX_EX);
        args.put("x-dead-letter-routing-key",DLX_RK);
        channel.queueDeclare(DELAY_Q, false, false, false, args);
        channel.queueBind(DELAY_Q, DELAY_EX, "");
    }
    //消费者监听 死信队列
    @Test
    public void consumerDelay() throws IOException, InterruptedException {
        //1.获取连接
        Channel channel = RabbitChannelManager.newChannel();
        //2.监听消费
        channel.basicConsume(
                DLX_Q,
                true,
                (consumerTag,message)->{
                    log.info("拿到延迟消息");
                },
                consumerTag -> {
                    log.info("消费者取消了监听");
                }
                );
        //3.阻塞线程,让消费者不停止
        Thread.sleep(10000000);
    }
    //生产者将消息组织后 发送到延迟交换机 路由到延迟队列
    //消息必须携带一个expiration属性
    @Test
    public void sendDelay() throws IOException {
        //1.获取连接
        Channel channel = RabbitChannelManager.newChannel();
        //2.组织消息
        byte[] body="延迟消息".getBytes();
        //3.消息属性携带expiration
        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
        builder.expiration("10000");//消息会延迟1分钟
        AMQP.BasicProperties properties = builder.build();
        //4.发送消息
        channel.basicPublish(DELAY_EX,"",properties,body);
        log.info("发送延迟消息成功");
    }


}
